package com.hp.pai.comm;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;

import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.log4j.Logger;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;

import com.alibaba.fastjson.JSON;

/**
 * profile从kafka获取进程端口信息 <Description> <br>
 * 
 * @author zgh<br>
 * @version 1.0<br>
 * @taskId <br>
 * @CreateDate 2016-3-24 <br>
 */
@SuppressWarnings("deprecation")
public class SynIProfileProcess implements Job
{

    private static Logger logger = Logger.getLogger(SynIProfileProcess.class);

    private SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    // 大数据平台接受数据的地址
    private String AIXURLSTR = "http://10.1.217.83:81/persist/pai_process_aix";

    private String LINUXURLSTR = "http://10.1.217.83:81/persist/pai_process_linux";

    private String WINURLSTR = "http://10.1.217.83:81/persist/pai_process_win";

    // 一次发送1000条数据
    private int cmtCount = 1000;

    final static String[] topics = { "app.process", "app.netconn",
            "app.linux.process", "app.linux.netconn", "app.win.process",
            "app.win.netconn" };

    // 各类map数据需要的字段名称
    final static String[] aixNet = { "Collect_time", "Serial_number",
            "Lpar_id", "Host_name", "Process_pid", "Local_port", "Local_ip",
            "Foreign_ip", "Foreign_port", "Process_name", "FILE_PATH",
            "ProcessUser" };

    final static String[] aixProcess = { "Collect_time", "FILE_PATH",
            "Process_pid", "Process_name", "ProcessUser" };

    final static String[] linuxNet = { "Collect_time", "Serial_number",
            "Lpar_id", "Host_name", "Process_pid", "Local_port", "Local_ip",
            "Foreign_ip", "Foreign_port", "Process_name", "FILE_PATH",
            "ProcessUser" };

    final static String[] linuxProcess = { "Collect_time", "FILE_PATH",
            "Process_pid", "Process_name", "ProcessUser" };

    final static String[] winNet = { "Collect_time", "Serial_number",
            "Lpar_id", "Host_name", "Process_pid", "Local_port", "Local_ip",
            "Foreign_ip", "Foreign_port", "Process_name", "FILE_PATH",
            "ProcessUser" };

    final static String[] winProcess = { "Collect_time", "FILE_PATH",
            "Process_pid", "Process_name", "ProcessUser" };

    public SynIProfileProcess()
    {
    }

    /**
     * Description: <br>
     * 
     * @author zgh<br>
     * @taskId <br>
     * @param arg0
     * @throws JobExecutionException <br>
     */
    @Override
    public void execute(JobExecutionContext arg0) throws JobExecutionException
    {
        try
        {
            logger.info("profileInter-SynIProfileProcess execute start......");
            long start = System.currentTimeMillis();
            synIProfileProcess();
            logger.info("SynIProfileProcess spent time detail:\n"
                    + (System.currentTimeMillis() - start) / 1000 + "s");
            logger.info("profileInter-SynIProfileProcess execute success......");
        }
        catch (Exception e)
        {

        }
    }

    @SuppressWarnings("unchecked")
    public void synIProfileProcess() throws IOException, Exception
    {
        Properties props = new Properties();
        // zookeeper
        props.put("zookeeper.connect", "10.1.250.4:2181,10.1.250.5:2181");
        props.put("zookeeper.session.timeout.ms", "30000");
        props.put("zookeeper.connection.timeout.ms", "40000");
        props.put("group.id", "g1");
        // props.put("auto.commit.enable", "false");

        /*
         * 由于数据推送是持续不断，所有需要我们自己判断一份完整数据的收集时间,此处按Collect_time收集2分钟内的数据，
         * 因为源数据是一分钟一次
         */
        Date startT = null;// 第一条数据的收集时间
        Date endT = null;// 一份完整数据收集完成的结束时间

        /******************** AIX **************************/
        // List<Map<String, String>> aixPortList = new ArrayList<Map<String,
        // String>>();// 端口
        List<Map<String, Object>> aixPortListWithKey = new ArrayList<Map<String, Object>>();// 推送到大数据平台的包括key的JSON数据
        Map<String, Map<String, String>> aixProMap = new HashMap<String, Map<String, String>>();// 进程的键值对,Process_pid:map
        ConsumerConnector consumer = Consumer
                .createJavaConsumerConnector(new ConsumerConfig(props));

        try
        {
            // 进程
            Map<String, Integer> map = new HashMap<String, Integer>();
            map.put(topics[0], 1);
            KafkaStream<byte[], byte[]> stream = consumer
                    .createMessageStreams(map).get(topics[0]).get(0);

            for (MessageAndMetadata<byte[], byte[]> msgAndMetadata : stream)
            {
                byte[] msg = msgAndMetadata.message();
                Object obj = JSON.parse(new String(msg, "utf8"));
                Map<String, String> source = (Map<String, String>) obj;
                if (source != null)
                {
                    Map<String, String> bean = getMapInfo0(source);
                    if (bean == null)
                    {
                        continue;
                    }
                    aixProMap.put(bean.get(aixProcess[2]), bean);
                    if (startT == null)
                    {
                        startT = this.getDateFromStr(bean.get(aixProcess[0]));
                        endT = this.caluteEndTime(startT);// 向后2分钟
                        continue;
                    }

                    // 如果Collect_time超过endT则结束
                    Date thisTime = this
                            .getDateFromStr(bean.get(aixProcess[0]));
                    if (thisTime != null && thisTime.after(endT))
                    {
                        break;
                    }
                }

            }
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            consumer.shutdown();
        }

        // 端口
        ConsumerConnector consumer1 = Consumer
                .createJavaConsumerConnector(new ConsumerConfig(props));
        try
        {
            Map<String, Integer> map1 = new HashMap<String, Integer>();
            map1.put(topics[1], 1);
            KafkaStream<byte[], byte[]> stream1 = consumer1
                    .createMessageStreams(map1).get(topics[1]).get(0);

            startT = null;// 第一条数据的收集时间
            endT = null;// 一份完整数据收集完成的结束时间
            for (MessageAndMetadata<byte[], byte[]> msgAndMetadata : stream1)
            {
                byte[] msg = msgAndMetadata.message();
                Object obj = JSON.parse(new String(msg, "utf8"));
                Map<String, String> source = (Map<String, String>) obj;
                if (source != null)
                {
                    Map<String, String> bean = getMapInfo1(source, aixProMap);
                    if (bean == null)
                    {
                        continue;
                    }
                    // aixPortList.add(bean);
                    String dataKey = this.getKeyVal(bean); // TODO 设计key值
                    Map<String, Object> json = new HashMap<String, Object>();
                    json.put("key", dataKey);
                    json.put("metainfo", bean);
                    aixPortListWithKey.add(json);
                    // 每一千条发送一次，然后clear
                    if (aixPortListWithKey.size() == cmtCount)
                    {
                        this.sendJsonInfo(aixPortListWithKey, AIXURLSTR);
                        aixPortListWithKey.clear();
                    }

                    if (startT == null)
                    {
                        startT = this.getDateFromStr(bean.get(aixNet[0]));
                        endT = this.caluteEndTime(startT);// 向后2分钟
                        continue;
                    }

                    // 如果Collect_time超过endT则结束
                    Date thisTime = this.getDateFromStr(bean.get(aixNet[0]));
                    if (thisTime != null && thisTime.after(endT))
                    {
                        break;
                    }
                }
            }
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            consumer1.shutdown();
        }

        // // 发送数据到大数据平台
        // List<Map<String, Object>> eachMaps = new ArrayList<Map<String,
        // Object>>();
        // for (int i = 0; i < aixPortListWithKey.size(); i++)
        // {
        // eachMaps.add(aixPortListWithKey.get(i));
        // if ((i + 1) % cmtCount == 0)
        // {
        // sendJsonInfo(eachMaps, AIXURLSTR);
        // eachMaps.clear();
        // }
        // }
        // if (eachMaps.size() > 0)
        // {
        // sendJsonInfo(eachMaps, AIXURLSTR);
        // }
        // 发送数据到大数据平台
        if (aixPortListWithKey.size() > 0)
        {
            sendJsonInfo(aixPortListWithKey, AIXURLSTR);
        }

        /***************************** LINUX **************************/
        // List<Map<String, String>> linuxPortList = new ArrayList<Map<String,
        // String>>();// 端口
        List<Map<String, Object>> linuxPortListWithKey = new ArrayList<Map<String, Object>>();// 推送到大数据平台的包括key的JSON数据
        Map<String, Map<String, String>> linuxProMap = new HashMap<String, Map<String, String>>();// 进程的键值对
                                                                                                  // Process_pid:map
        // 进程
        ConsumerConnector consumer2 = Consumer
                .createJavaConsumerConnector(new ConsumerConfig(props));
        try
        {
            Map<String, Integer> map2 = new HashMap<String, Integer>();
            map2.put(topics[2], 1);
            KafkaStream<byte[], byte[]> stream2 = consumer2
                    .createMessageStreams(map2).get(topics[2]).get(0);

            /*
             * 由于数据推送是持续不断，所有需要我们自己判断一份完整数据的收集时间,此处按Collect_time收集2分钟内的数据，
             * 因为源数据是一分钟一次
             */
            startT = null;// 第一条数据的收集时间
            endT = null;// 一份完整数据收集完成的结束时间
            for (MessageAndMetadata<byte[], byte[]> msgAndMetadata : stream2)
            {
                byte[] msg = msgAndMetadata.message();
                Object obj = JSON.parse(new String(msg, "utf8"));
                Map<String, String> source = (Map<String, String>) obj;
                if (source != null)
                {
                    Map<String, String> bean = getMapInfo2(source);
                    if (bean == null)
                    {
                        continue;
                    }
                    linuxProMap.put(bean.get(linuxProcess[2]), bean);
                    if (startT == null)
                    {
                        startT = this.getDateFromStr(bean.get(linuxProcess[0]));
                        endT = this.caluteEndTime(startT);// 向后2分钟
                        continue;
                    }

                    // 如果Collect_time超过endT则结束
                    Date thisTime = this.getDateFromStr(bean
                            .get(linuxProcess[0]));
                    if (thisTime != null && thisTime.after(endT))
                    {
                        break;
                    }
                }

            }
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            consumer2.shutdown();
        }

        // 端口
        ConsumerConnector consumer3 = Consumer
                .createJavaConsumerConnector(new ConsumerConfig(props));

        try
        {
            Map<String, Integer> map3 = new HashMap<String, Integer>();
            map3.put(topics[3], 1);
            KafkaStream<byte[], byte[]> stream3 = consumer3
                    .createMessageStreams(map3).get(topics[3]).get(0);

            startT = null;// 第一条数据的收集时间
            endT = null;// 一份完整数据收集完成的结束时间
            for (MessageAndMetadata<byte[], byte[]> msgAndMetadata : stream3)
            {
                byte[] msg = msgAndMetadata.message();
                Object obj = JSON.parse(new String(msg, "utf8"));
                Map<String, String> source = (Map<String, String>) obj;
                if (source != null)
                {
                    Map<String, String> bean = getMapInfo3(source, linuxProMap);
                    if (bean == null)
                    {
                        continue;
                    }
                    // linuxPortList.add(bean);
                    String dataKey = this.getKeyVal(bean); // TODO 设计key值
                    Map<String, Object> json = new HashMap<String, Object>();
                    json.put("key", dataKey);
                    json.put("metainfo", bean);
                    linuxPortListWithKey.add(json);
                    // 每一千条发送一次，然后clear
                    if (linuxPortListWithKey.size() == cmtCount)
                    {
                        this.sendJsonInfo(linuxPortListWithKey, LINUXURLSTR);
                        linuxPortListWithKey.clear();
                    }

                    if (startT == null)
                    {
                        startT = this.getDateFromStr(bean.get(linuxNet[0]));
                        endT = this.caluteEndTime(startT);// 向后2分钟
                        continue;
                    }

                    // 如果Collect_time超过endT则结束
                    Date thisTime = this.getDateFromStr(bean.get(linuxNet[0]));
                    if (thisTime != null && thisTime.after(endT))
                    {
                        break;
                    }
                }
            }
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            consumer3.shutdown();
        }

        // 发送数据到大数据平台
        if (linuxPortListWithKey.size() > 0)
        {
            sendJsonInfo(linuxPortListWithKey, LINUXURLSTR);
        }

        /****************************** WIN **************************/
        // List<Map<String, String>> winPortList = new ArrayList<Map<String,
        // String>>();// 端口
        List<Map<String, Object>> winPortListWithKey = new ArrayList<Map<String, Object>>();// 推送到大数据平台的包括key的JSON数据
        Map<String, Map<String, String>> winProMap = new HashMap<String, Map<String, String>>();// 进程的键值对
                                                                                                // Process_pid:map
        // 进程
        ConsumerConnector consumer4 = Consumer
                .createJavaConsumerConnector(new ConsumerConfig(props));

        try
        {
            Map<String, Integer> map4 = new HashMap<String, Integer>();
            map4.put(topics[4], 1);
            KafkaStream<byte[], byte[]> stream4 = consumer4
                    .createMessageStreams(map4).get(topics[4]).get(0);

            /*
             * 由于数据推送是持续不断，所有需要我们自己判断一份完整数据的收集时间,此处按Collect_time收集2分钟内的数据，
             * 因为源数据是一分钟一次
             */
            startT = null;// 第一条数据的收集时间
            endT = null;// 一份完整数据收集完成的结束时间
            for (MessageAndMetadata<byte[], byte[]> msgAndMetadata : stream4)
            {
                byte[] msg = msgAndMetadata.message();
                Object obj = JSON.parse(new String(msg, "utf8"));
                Map<String, String> source = (Map<String, String>) obj;
                if (source != null)
                {
                    Map<String, String> bean = getMapInfo4(source);
                    if (bean == null)
                    {
                        continue;
                    }
                    winProMap.put(bean.get(winProcess[2]), bean);
                    if (startT == null)
                    {
                        startT = this.getDateFromStr(bean.get(winProcess[0]));
                        endT = this.caluteEndTime(startT);// 向后2分钟
                        continue;
                    }

                    // 如果Collect_time超过endT则结束
                    Date thisTime = this
                            .getDateFromStr(bean.get(winProcess[0]));
                    if (thisTime != null && thisTime.after(endT))
                    {
                        break;
                    }
                }

            }
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            consumer4.shutdown();
        }

        // 端口
        ConsumerConnector consumer5 = Consumer
                .createJavaConsumerConnector(new ConsumerConfig(props));
        try
        {
            Map<String, Integer> map5 = new HashMap<String, Integer>();
            map5.put(topics[5], 1);
            KafkaStream<byte[], byte[]> stream5 = consumer5
                    .createMessageStreams(map5).get(topics[5]).get(0);

            startT = null;// 第一条数据的收集时间
            endT = null;// 一份完整数据收集完成的结束时间
            for (MessageAndMetadata<byte[], byte[]> msgAndMetadata : stream5)
            {
                byte[] msg = msgAndMetadata.message();
                Object obj = JSON.parse(new String(msg, "utf8"));
                Map<String, String> source = (Map<String, String>) obj;
                if (source != null)
                {
                    Map<String, String> bean = getMapInfo5(source, winProMap);
                    if (bean == null)
                    {
                        continue;
                    }
                    // winPortList.add(bean);
                    String dataKey = this.getKeyVal(bean); // TODO 设计key值
                    Map<String, Object> json = new HashMap<String, Object>();
                    json.put("key", dataKey);
                    json.put("metainfo", bean);
                    winPortListWithKey.add(json);
                    // 每一千条发送一次，然后clear
                    if (winPortListWithKey.size() == cmtCount)
                    {
                        this.sendJsonInfo(winPortListWithKey, WINURLSTR);
                        winPortListWithKey.clear();
                    }

                    if (startT == null)
                    {
                        startT = this.getDateFromStr(bean.get(winNet[0]));
                        endT = this.caluteEndTime(startT);// 向后2分钟
                        continue;
                    }

                    // 如果Collect_time超过endT则结束
                    Date thisTime = this.getDateFromStr(bean.get(winNet[0]));
                    if (thisTime != null && thisTime.after(endT))
                    {
                        break;
                    }
                }
            }

        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        finally
        {
            consumer5.shutdown();
        }

        // 发送数据到大数据平台
        if (winPortListWithKey.size() > 0)
        {
            sendJsonInfo(winPortListWithKey, WINURLSTR);
        }
    }

    /**
     * 存储到大数据平台的key值设计 Description: <br>
     * 
     * @author zgh<br>
     * @taskId <br>
     * @param bean
     * @return <br>
     */
    private String getKeyVal(Map<String, String> bean)
    {
        String key = null;
        if (bean != null)
        {
            key = bean.get("Process_pid") + "A" + bean.get("Local_port");
        }
        return key;
    }

    /**
     * 获取2分钟后的时间 Description: <br>
     * 
     * @author zgh<br>
     * @taskId <br>
     * @param startT
     * @return <br>
     */
    private Date caluteEndTime(Date startT)
    {
        if (startT == null)
        {
            return null;
        }
        Calendar time = Calendar.getInstance();
        time.setTime(startT);
        time.add(Calendar.MINUTE, 2);

        return time.getTime();
    }

    /**
     * AIX进程 Description: <br>
     * 
     * @author zgh<br>
     * @taskId <br>
     * @param source
     * @return <br>
     */
    public Map<String, String> getMapInfo0(Map<String, String> source)
    {
        Map<String, String> ret = null;
        if (source != null)
        {
            ret = new HashMap<String, String>();
            ret.put(aixProcess[1], source.get("FILE_PATH"));

            String logStr = source.get("LOG");
            if (logStr != null)
            {
                String[] arr = logStr.split("\\|");
                if (arr.length > 10)
                {
                    ret.put(aixProcess[0], arr[0].trim());// 数据收集时间
                    ret.put(aixProcess[2], arr[6].trim());// Process_pid唯一标识，关联端口的Process_pid
                    ret.put(aixProcess[3], arr[10].trim());// 进程名
                    ret.put(aixProcess[4], arr[5].trim());// 用户
                }
                else
                {
                    return null;
                }
            }
        }
        return ret;
    }

    /**
     * AIX端口 Description: <br>
     * 
     * @author zgh<br>
     * @taskId <br>
     * @param source
     * @param aixProMap
     * @return <br>
     */
    public Map<String, String> getMapInfo1(Map<String, String> source,
            Map<String, Map<String, String>> aixProMap)
    {
        Map<String, String> ret = null;
        if (source != null)
        {
            ret = new HashMap<String, String>();
            String logStr = source.get("LOG");
            if (logStr != null)
            {
                String[] arr = logStr.split("\\|");
                if (arr.length > 10)
                {
                    ret.put(aixNet[0], arr[0].trim());// 数据收集时间
                    ret.put(aixNet[1], arr[1].trim());
                    ret.put(aixNet[2], arr[2].trim());
                    ret.put(aixNet[3], arr[4].trim());
                    ret.put(aixNet[4], arr[5].trim());// 关联进程的Process_pid
                    ret.put(aixNet[5], arr[7].trim());// 端口
                    ret.put(aixNet[6], arr[6].trim());//
                    ret.put(aixNet[7], arr[8].trim());//
                    ret.put(aixNet[8], arr[9].trim());//

                    if (aixProMap != null
                            && aixProMap.get(arr[5].trim()) != null)
                    {
                        // 关联的进程信息
                        ret.put(aixNet[9],
                                aixProMap.get(arr[5].trim()).get(aixProcess[3]));
                        ret.put(aixNet[10],
                                aixProMap.get(arr[5].trim()).get(aixProcess[1]));
                        ret.put(aixNet[11],
                                aixProMap.get(arr[5].trim()).get(aixProcess[4]));
                    }
                }
                else
                {
                    return null;
                }
            }
        }
        return ret;
    }

    /**
     * linux进程 Description: <br>
     * 
     * @author zgh<br>
     * @taskId <br>
     * @param source
     * @return <br>
     */
    public Map<String, String> getMapInfo2(Map<String, String> source)
    {
        Map<String, String> ret = null;
        if (source != null)
        {
            ret = new HashMap<String, String>();
            ret.put(linuxProcess[1], source.get("FILE_PATH"));

            String logStr = source.get("LOG");
            if (logStr != null)
            {
                String[] arr = logStr.split("\\|");
                if (arr.length > 10)
                {
                    ret.put(linuxProcess[0], arr[0].trim());// 数据收集时间
                    ret.put(linuxProcess[2], arr[6].trim());// Process_pid唯一标识，关联端口的Process_pid
                    ret.put(linuxProcess[3], arr[10].trim());// 进程名
                    ret.put(linuxProcess[4], arr[5].trim());// 用户
                }
                else
                {
                    return null;
                }
            }
        }
        return ret;
    }

    /**
     * linux端口 Description: <br>
     * 
     * @author zgh<br>
     * @taskId <br>
     * @param source
     * @param aixProMap
     * @return <br>
     */
    public Map<String, String> getMapInfo3(Map<String, String> source,
            Map<String, Map<String, String>> aixProMap)
    {
        Map<String, String> ret = null;
        if (source != null)
        {
            ret = new HashMap<String, String>();
            String logStr = source.get("LOG");
            if (logStr != null)
            {
                String[] arr = logStr.split("\\|");
                if (arr.length > 10)
                {
                    ret.put(linuxNet[0], arr[0].trim());// 数据收集时间
                    ret.put(linuxNet[1], arr[1].trim());
                    ret.put(linuxNet[2], arr[2].trim());
                    ret.put(linuxNet[3], arr[4].trim());
                    ret.put(linuxNet[4], arr[5].trim());// 关联进程的Process_pid
                    ret.put(linuxNet[5], arr[7].trim());// 端口
                    ret.put(linuxNet[6], arr[6].trim());//
                    ret.put(linuxNet[7], arr[8].trim());//
                    ret.put(linuxNet[8], arr[9].trim());//

                    if (aixProMap != null
                            && aixProMap.get(arr[5].trim()) != null)
                    {
                        // 关联的进程信息
                        ret.put(linuxNet[9],
                                aixProMap.get(arr[5].trim()).get(
                                        linuxProcess[3]));
                        ret.put(linuxNet[10],
                                aixProMap.get(arr[5].trim()).get(
                                        linuxProcess[1]));
                        ret.put(linuxNet[11],
                                aixProMap.get(arr[5].trim()).get(
                                        linuxProcess[4]));
                    }
                }
                else
                {
                    return null;
                }
            }
        }
        return ret;
    }

    /**
     * win进程 Description: <br>
     * 
     * @author zgh<br>
     * @taskId <br>
     * @param source
     * @return <br>
     */
    public Map<String, String> getMapInfo4(Map<String, String> source)
    {
        Map<String, String> ret = null;
        if (source != null)
        {
            ret = new HashMap<String, String>();
            ret.put(winProcess[1], source.get("FILE_PATH"));

            String logStr = source.get("LOG");
            if (logStr != null)
            {
                String[] arr = logStr.split("\\|");
                if (arr.length > 6)
                {
                    ret.put(winProcess[0], arr[0].trim());// 数据收集时间
                    ret.put(winProcess[2], arr[5].trim());// Process_pid唯一标识，关联端口的Process_pid
                    ret.put(winProcess[3], arr[6].trim());// 进程名
                }
                else
                {
                    return null;
                }
            }
        }
        return ret;
    }

    /**
     * win端口 Description: <br>
     * 
     * @author zgh<br>
     * @taskId <br>
     * @param source
     * @param aixProMap
     * @return <br>
     */
    public Map<String, String> getMapInfo5(Map<String, String> source,
            Map<String, Map<String, String>> aixProMap)
    {
        Map<String, String> ret = null;
        if (source != null)
        {
            ret = new HashMap<String, String>();
            String logStr = source.get("LOG");
            if (logStr != null)
            {
                String[] arr = logStr.split("\\|");
                if (arr.length > 11)
                {
                    ret.put(winNet[0], arr[0].trim());// 数据收集时间
                    ret.put(winNet[1], arr[1].trim());
                    ret.put(winNet[2], arr[2].trim());
                    ret.put(winNet[3], arr[4].trim());
                    ret.put(winNet[4], arr[11].trim());// 关联进程的Process_pid
                    ret.put(winNet[5], arr[7].trim());// 端口
                    ret.put(winNet[6], arr[6].trim());//
                    ret.put(winNet[7], arr[8].trim());//
                    ret.put(winNet[8], arr[9].trim());//

                    if (aixProMap != null
                            && aixProMap.get(arr[5].trim()) != null)
                    {
                        // 关联的进程信息
                        ret.put(winNet[9],
                                aixProMap.get(arr[5].trim()).get(winProcess[3]));
                        ret.put(winNet[10],
                                aixProMap.get(arr[5].trim()).get(winProcess[1]));
                    }
                }
                else
                {
                    return null;
                }
            }
        }
        return ret;
    }

    // public void sendJsonInfo(String JSONdata, String urlStr)
    // {
    // URL url = null;// 请求处理的Servlet
    //
    // ObjectOutputStream objOutputStrm = null;// 对象输出流
    //
    // HttpURLConnection httpUrlConnection = null;
    //
    // try
    // {
    // // url = new URL("http://10.1.217.83:81/persist/pai_process_aix");
    // url = new URL(urlStr);
    // httpUrlConnection = (HttpURLConnection) url.openConnection();
    // httpUrlConnection.setRequestMethod("POST");
    // httpUrlConnection.setRequestProperty("Content-Type",
    // "application/json");
    // // 发送POST请求必须设置如下两行
    // httpUrlConnection.setDoOutput(true);
    // httpUrlConnection.setDoInput(true);
    //
    // // 获取URLConnection对象对应的输出流
    // objOutputStrm = new ObjectOutputStream(
    // httpUrlConnection.getOutputStream());
    // objOutputStrm.writeObject(JSONdata);
    // objOutputStrm.flush();
    // objOutputStrm.close();
    //
    // httpUrlConnection.disconnect();
    // }
    // catch (Exception e)
    // {
    // // TODO Auto-generated catch block
    // e.printStackTrace();
    // }
    // }
    @SuppressWarnings({ "resource" })
    public void sendJsonInfo(List<Map<String, Object>> data, String urlStr)
    {
        logger.info("put数据到：" + urlStr);
        Map<String, Object> info = new HashMap<String, Object>();
        // 服务端json数据格式最外层为固定：data
        info.put("data", data);
        String JSONdata = JSON.toJSONString(info);
        try
        {
            DefaultHttpClient httpClient = new DefaultHttpClient();
            HttpPost httpPost = new HttpPost(urlStr);
            httpPost.setHeader("Content-Type", "application/json");
            StringEntity entity = new StringEntity(JSONdata, "UTF-8");
            entity.setContentEncoding("UTF-8");// 解决中文乱码
            entity.setContentType("application/json");
            httpPost.setEntity(entity);

            // 通过访问调用，获取到返回的response
            HttpResponse response = httpClient.execute(httpPost);
            // String resData = EntityUtils
            // .toString(response.getEntity(), "UTF-8");
            // System.out.println(resData);
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }

    /**
     * 字符转日期 Description: <br>
     * 
     * @author zgh<br>
     * @taskId <br>
     * @param str
     * @return <br>
     */
    public Date getDateFromStr(String str)
    {
        Date ret = null;

        try
        {
            ret = df.parse(str);
        }
        catch (ParseException e)
        {
            e.printStackTrace();
        }
        return ret;
    }

}
